package FinalProject

import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.clients.consumer.ConsumerConfig
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.streaming.dstream.InputDStream
import org.apache.spark.streaming.kafka010._
import scala.collection.mutable.HashMap

object KafkaSparkStreaming {

  def main(args: Array[String]): Unit = {
    // 配置Spark Streaming相关参数
    val sparkConf = new SparkConf().setAppName("KafkaSparkStreaming").setMaster("local[*]")
    val ssc = new StreamingContext(sparkConf, Seconds(2))
    ssc.sparkContext.setLogLevel("error")

    // 配置Kafka相关参数，添加从最早偏移量开始消费的设置
    val kafkaParams = Map[String, Object](
      ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG -> "8.152.219.6:9092",
      ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG -> classOf[StringDeserializer],
      ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG -> classOf[StringDeserializer],
      ConsumerConfig.GROUP_ID_CONFIG -> "niit12345",
      ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG -> (false: java.lang.Boolean),
      // 设置从最早的偏移量开始消费，这样就能获取到之前的数据
      ConsumerConfig.AUTO_OFFSET_RESET_CONFIG -> "earliest"
    )

    val topics = Array("stuInfoTest")

    // 创建基于Kafka的输入DStream
    val stream: InputDStream[ConsumerRecord[String, String]] = KafkaUtils.createDirectStream[String, String](
      ssc,
      LocationStrategies.PreferConsistent,
      ConsumerStrategies.Subscribe[String, String](topics.toList, kafkaParams)
    )

    // 用于存储统计结果的可变映射，键为 (性别, 在籍状态) 二元组，值为数量
    val statsMap = new HashMap[(String, String), Long]()

    // 处理每条消息进行统计，并实时打印当前统计情况（可按需调整输出逻辑）
    stream.foreachRDD { rdd =>
      rdd.foreach { record =>
        val studentInfo = record.value().split("\t")
        val gender = studentInfo(2).trim
        val status = studentInfo(6).trim
        val key = (gender, status)
        statsMap(key) = statsMap.getOrElse(key, 0L) + 1L
        // 实时打印当前统计情况
        println(s"当前已接收消息，性别: $gender, 在籍状态: $status, 数量: ${statsMap(key)}")
      }
    }

    ssc.start()
    ssc.awaitTermination()
  }
}